package practice.phase03.module04.zookeeper;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.Properties;

/**
 * @author Alex Zhang
 */
public class ZkTest {
    private ZkClient zk = new ZkClient("node1:2181,node2:2181,node3:2181");

    @Test
    void createNode() {
        String path = "/zkclient";
        String helloZkClient = zk.create("/zkclient", "hello_zk_client", CreateMode.PERSISTENT_SEQUENTIAL);
        // 级联创建
        zk.createPersistent("/1/2", true);
        Assertions.assertTrue(helloZkClient.startsWith(path));
    }

    @Test
    void deleteNode() {
        String path = "/zkclient0000000002";
        Assertions.assertTrue(zk.delete(path));
    }

    @Test
    void deleteRecursive() {
        String path = "/zkclient0000000002";
        Assertions.assertTrue(zk.deleteRecursive(path));
    }

    @Test
    void watcherChildrenChange() throws InterruptedException {
        zk.subscribeChildChanges("/client", (s, children) -> System.out.println("watcher triggered: " + s + ", children: " + children));
        zk.createPersistent("/client");
        Thread.sleep(1000);
        zk.createPersistent("/client/child");
        zk.writeData("/client/child", "hello");
        Thread.sleep(1000);
        zk.delete("/client/child");
        Thread.sleep(1000);
        zk.delete("/client");

        // 用来模拟在命令行操作也会触发这里创建的watcher
//        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    void watcherDataChange() {
        zk.subscribeDataChanges("/datachange", new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) {
                System.out.println(s + ", data: " + o);
            }

            @Override
            public void handleDataDeleted(String s) {
                System.out.println(s);
            }
        });

    }

    @Test
    void mySql() {
        final String mysqlZkPath = "/mysql";
        if (zk.exists(mysqlZkPath)) {
            Properties p = zk.readData(mysqlZkPath);
            p.setProperty("password", "dont4gett");
            zk.writeData(mysqlZkPath, p);
            System.out.println(p);
        }
    }

    private void deleteRecursively(ZkClient zk, String path) {
        if(zk.exists(path)) {
            zk.deleteRecursive(path);
        }
    }

    @Test
    void recursivelyDelete() {
        String[] servers = new String[] {"node1:2181", "node2:2181", "node3:2181"};
        String[] nodes = new String[] {"/kafka"};
        for(String server: servers) {
            ZkClient zk = new ZkClient(server);
            for(String node: nodes) {
                deleteRecursively(zk, node);
            }
        }
    }
}
